package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/sirupsen/logrus"
	"github.com/spf13/viper"
)

func main() {
	// 配置 viper
	viper.SetConfigName("config")
	viper.SetConfigType("yaml")
	viper.AddConfigPath("..")

	// 读取配置文件
	if err := viper.ReadInConfig(); err != nil {
		var configFileNotFoundError viper.ConfigFileNotFoundError
		if errors.As(err, &configFileNotFoundError) {
			log.Fatalf("Config file not found; %v", err)
		}
	}

	// 从配置文件获取 RocketMQ 配置
	groupName := viper.GetString("rocketmq.group_name")
	logrus.Info(groupName)
	nameserver := viper.GetString("rocketmq.name_server")
	logrus.Info(nameserver)
	consumerTopic := viper.GetString("rocketmq.consumer_topic")
	logrus.Info(consumerTopic)

	nameservers := viper.GetStringSlice("rocketmq.name_server")
	// 打印 nameserver 的值
	fmt.Println("Nameservers:")
	for _, ns := range nameservers {
		fmt.Println(ns)
	}

	var c rocketmq.PushConsumer
	var err error
	for {
		// 创建一个新的消费者实例
		c, err = rocketmq.NewPushConsumer(
			consumer.WithGroupName(groupName),
			consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{nameserver})),
		)
		if err == nil {
			log.Println("消费者创建成功")
			break
		}
		log.Printf("创建消费者失败: %s，将在 3 秒后重试", err.Error())
		time.Sleep(3 * time.Second)
	}

	// 订阅主题
	msgChan := make(chan *primitive.MessageExt, 100) // 创建消息通道
	var wg sync.WaitGroup

	// 启动 10 个并行的消费者 goroutine
	for i := 0; i < 10; i++ {

		wg.Add(1)
		go func() {
			defer wg.Done()
			for msg := range msgChan {
				fmt.Printf("收到消息: %s\n", string(msg.Body))
			}
		}()
	}

	err = c.Subscribe(consumerTopic, consumer.MessageSelector{}, func(ctx context.Context,
		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			msgChan <- msg // 将消息发送到通道
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		log.Fatalf("订阅主题失败: %s", err.Error())
	}

	// 启动消费者，添加重试逻辑
	var startErr error
	for {
		startErr = c.Start()
		if startErr == nil {
			log.Println("消费者启动成功")
			break
		}
		log.Printf("启动消费者失败: %s，将在 3 秒后重试", startErr.Error())
		time.Sleep(3 * time.Second)
	}

	// 保持程序运行
	select {}
}
